package com.star.sync.elasticsearch.service.impl;

import com.alibaba.fastjson.JSON;
import com.star.sync.elasticsearch.service.ElasticsearchService;
import com.star.sync.elasticsearch.util.JsonUtil;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.rest.RestStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.List;
import java.util.Map;

/**
 * @author <a href="mailto:wangchao.star@gmail.com">wangchao</a>
 * @version 1.0
 * @since 2017-08-27 12:38:00
 */
@Service
public class ElasticsearchServiceImpl implements ElasticsearchService {
    private static final Logger logger = LoggerFactory.getLogger(ElasticsearchServiceImpl.class);

    @Resource
    private RestHighLevelClient restHighLevelClient;


    @Override
    public void insertById(String index, String type, String id, Map<String, Object> dataMap) {
        IndexRequest request = new IndexRequest(index, "_doc", id);
        request.source(JSON.toJSONString(dataMap), XContentType.JSON);
        IndexResponse indexResponse = null;
        try {
            // 同步方式
            indexResponse = restHighLevelClient.index(request, RequestOptions.DEFAULT);
        } catch (ElasticsearchException e) {
            // 捕获，并处理异常
            //判断是否版本冲突、create但文档已存在冲突
            if (e.status() == RestStatus.CONFLICT) {
                System.out.println("冲突了，请在此写冲突处理逻辑！" + e.getDetailedMessage());
            }
            e.printStackTrace();
        }catch (IOException e){

            e.printStackTrace();
        }
    }

    @Override
    public void batchInsertById(String index, String type, List<Map<String, Object>> idDataMap) {
        BulkRequest bulkRequest = new BulkRequest();
        try {
            for (Map<String, Object> a : idDataMap) {
                bulkRequest.add(new IndexRequest(index, type, String.valueOf(a.get("id")))
                        .source(JSON.toJSONString(a)));
            }
            BulkResponse bulkResponse = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (bulkResponse.hasFailures()) {
                logger.error("elasticsearch批量插入错误, index=" + index + ", type=" + type + ", data=" + JsonUtil.toJson(idDataMap) + ", cause:" + bulkResponse.buildFailureMessage());
            }
        } catch (Exception e) {
            logger.error("elasticsearch批量插入错误, index=" + index + ", type=" + type + ", data=" + JsonUtil.toJson(idDataMap), e);
        }
    }

    @Override
    public void update(String index, String type, String id, Map<String, Object> dataMap) {
        this.insertById(index, type, id, dataMap);
    }

    @Override
    public void deleteById(String index, String type, String id) throws IOException {
        DeleteRequest request = new DeleteRequest(index, "_doc", id);
        restHighLevelClient.delete(request,RequestOptions.DEFAULT);
    }
}
